package com.example.rocketmq.demo.producer;

import com.example.rocketmq.demo.config.RocketMQConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * 基于注解方式的消息生产者
 */
@Slf4j
@Service
public class AnnotationProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 同步发送普通消息
     *
     * @param content 消息内容
     * @return 发送结果
     */
    public SendResult sendSync(String content) {
        // 构建消息
        Message<String> message = MessageBuilder.withPayload(content)
                .setHeader(RocketMQHeaders.KEYS, UUID.randomUUID().toString())
                .build();
        
        // 同步发送
        SendResult sendResult = rocketMQTemplate.syncSend(RocketMQConstant.TOPIC_NORMAL, message);
        log.info("注解方式-同步发送消息结果: {}", sendResult);
        return sendResult;
    }

    /**
     * 异步发送普通消息
     *
     * @param content 消息内容
     */
    public void sendAsync(String content) {
        // 构建消息
        Message<String> message = MessageBuilder.withPayload(content)
                .setHeader(RocketMQHeaders.KEYS, UUID.randomUUID().toString())
                .build();
        
        // 异步发送
        rocketMQTemplate.asyncSend(RocketMQConstant.TOPIC_NORMAL, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("注解方式-异步发送消息成功: {}", sendResult);
            }

            @Override
            public void onException(Throwable e) {
                log.error("注解方式-异步发送消息失败", e);
            }
        });
    }

    /**
     * 单向发送消息（不关心发送结果）
     *
     * @param content 消息内容
     */
    public void sendOneWay(String content) {
        // 构建消息
        Message<String> message = MessageBuilder.withPayload(content)
                .setHeader(RocketMQHeaders.KEYS, UUID.randomUUID().toString())
                .build();
        
        // 单向发送
        rocketMQTemplate.sendOneWay(RocketMQConstant.TOPIC_NORMAL, message);
        log.info("注解方式-单向发送消息完成");
    }

    /**
     * 发送延时消息
     *
     * @param content    消息内容
     * @param delayLevel 延时等级 (1-18)
     * @return 发送结果
     */
    public SendResult sendDelay(String content, int delayLevel) {
        // 构建消息
        Message<String> message = MessageBuilder.withPayload(content)
                .setHeader(RocketMQHeaders.KEYS, UUID.randomUUID().toString())
                .build();
        
        // 发送延时消息
        SendResult sendResult = rocketMQTemplate.syncSend(
                RocketMQConstant.TOPIC_DELAY,
                message,
                3000, // 超时时间
                delayLevel // 延时等级
        );
        log.info("注解方式-发送延时消息结果: {}", sendResult);
        return sendResult;
    }

    /**
     * 发送顺序消息
     *
     * @param content 消息内容
     * @param orderId 订单ID（用于消息分区）
     * @return 发送结果
     */
    public SendResult sendOrderly(String content, String orderId) {
        // 构建消息
        Message<String> message = MessageBuilder.withPayload(content)
                .setHeader(RocketMQHeaders.KEYS, UUID.randomUUID().toString())
                .build();
        
        // 发送顺序消息
        SendResult sendResult = rocketMQTemplate.syncSendOrderly(
                RocketMQConstant.TOPIC_ORDER,
                message,
                orderId // 分区键
        );
        log.info("注解方式-发送顺序消息结果: {}", sendResult);
        return sendResult;
    }

    /**
     * 发送带标签的过滤消息
     *
     * @param content 消息内容
     * @param tag     消息标签
     * @return 发送结果
     */
    public SendResult sendWithTag(String content, String tag) {
        // 构建消息
        Message<String> message = MessageBuilder.withPayload(content)
                .setHeader(RocketMQHeaders.KEYS, UUID.randomUUID().toString())
                .build();
        
        // 发送带标签的消息
        SendResult sendResult = rocketMQTemplate.syncSend(
                RocketMQConstant.TOPIC_FILTER + ":" + tag,
                message
        );
        log.info("注解方式-发送带标签的过滤消息结果: {}", sendResult);
        return sendResult;
    }

    /**
     * 发送带属性的SQL过滤消息
     *
     * @param content 消息内容
     * @param age     年龄属性
     * @return 发送结果
     */
    public SendResult sendWithProperty(String content, int age) {
        // 构建消息
        Message<String> message = MessageBuilder.withPayload(content)
                .setHeader(RocketMQHeaders.KEYS, UUID.randomUUID().toString())
                .setHeader("age", age)
                .build();
        
        // 发送带属性的消息
        SendResult sendResult = rocketMQTemplate.syncSend(RocketMQConstant.TOPIC_FILTER, message);
        log.info("注解方式-发送带属性的SQL过滤消息结果: {}", sendResult);
        return sendResult;
    }

    /**
     * 发送事务消息
     *
     * @param content 消息内容
     * @param status  事务状态（用于模拟本地事务执行结果）
     * @return 发送结果
     */
    public TransactionSendResult sendTransaction(String content, String status) {
        // 构建消息
        Message<String> message = MessageBuilder.withPayload(content)
                .setHeader(RocketMQHeaders.KEYS, UUID.randomUUID().toString())
                .build();
        
        // 发送事务消息
        TransactionSendResult sendResult = rocketMQTemplate.sendMessageInTransaction(
                RocketMQConstant.TOPIC_TRANSACTION,
                message,
                status // 传递给本地事务执行器的参数
        );
        log.info("注解方式-发送事务消息结果: {}", sendResult);
        return sendResult;
    }
}